#include <sstream>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/time.h>
#include <sys/stat.h>
#include "ConnManager.h"
#include "SocketConn.h"
#include "HttpConn.h"
#include "Http2Conn.h"
#include "NantxConn.h"
#include "configParser.h"
#include "Debug.h"

#define CONNMANAGER_MAX_PARTITION      255
#define CONNMANAGER_MIN_PARTITION      1

using namespace netproxy;
bool ConnManager:: Initialize(const char *conf)
{
	struct stat st;
	if (lstat(conf, &st) != 0)
	{
		cerr<<__FILE__<<":"<<__func__<<":"<<__LINE__<<" file '"<<conf<<"' is not exist!"<<endl;
		return false;
	}

	configParser cfg;
	map <string, string > g_Confs;
	g_Confs = cfg.Parser(conf);
	if(g_Confs.size() <= 0)
	{
		cerr<<"error to explain conf :"<< conf <<endl;
		return false;
	}
        string stemp;
	if( g_Confs.find("max") != g_Confs.end())
	{
      	     stemp  = g_Confs["max"];
             m_iPartition = atoi(stemp.c_str());
             if(m_iPartition < CONNMANAGER_MIN_PARTITION || m_iPartition > CONNMANAGER_MAX_PARTITION)
             {
                ERRO<<" max is error!!! "<<endl;
                return false;
             } 
        }
        else
           return false;
/*
        int iMTU = 1448;
	if( g_Confs.find("MTU") != g_Confs.end())
	{
      	     stemp  = g_Confs["MTU"];
             iMTU = atoi(stemp.c_str());
        }
        DEBUG2("NOW configure setting MTU = ", iMTU);
*/

	int *pIA = new int[ m_iPartition + 1];
	memset(pIA, -1, m_iPartition + 1);
	for(int i=CONNMANAGER_MIN_PARTITION; i<= m_iPartition ; i++ )
	{
		ostringstream os;
		os << i;
		if( g_Confs.find(os.str()) != g_Confs.end())
		{
			stemp = g_Confs[os.str()];  //host:port:rtable,rtable,rtable
			size_t idx = stemp.find(':', 0);
			if(idx> 0 && idx < stemp.length())
			{
				string shost = stemp.substr(0, idx); //host 
				string sport, sbackup;
				trim(shost);
				stemp = stemp.substr(idx + 1, stemp.length()- idx -1);
				idx = stemp.find(':', 0);
				if(idx>0 && idx < stemp.length())
				{
					sport = stemp.substr(0, idx); // port
					sbackup = stemp.substr(idx + 1, stemp.length()- idx -1); //rtable, such as: 2,3,4
				}
				if(pIA[i] != i)
				{
					pIA[i] = i;
					ConnStruct  *pSt = new ConnStruct;
					if(m_iType == 0) 
						pSt->pCon = new SocketConn(shost.c_str(),atoi(sport.c_str()),i, sbackup.c_str()); 
					else if(m_iType == 1)
						pSt->pCon = new HttpConn(shost.c_str(),atoi(sport.c_str()),i, sbackup.c_str());  
        				else if(m_iType == 2)
						pSt->pCon = new Http2Conn(shost.c_str(),atoi(sport.c_str()),i, sbackup.c_str());
					else if(m_iType == 3)
						pSt->pCon = new NantxConn(shost.c_str(),atoi(sport.c_str()),i, sbackup.c_str());
					else
					{
                                             ERRO <<"NO type = "<< m_iType <<" exit;"<<endl;
                                             exit(0);
					}
//  pSt->pCon->SetMTUSize(iMTU);     
                 
#ifdef _THREAD_SAFE
					pthread_mutex_init(&(pSt->lock), NULL);
#endif
					m_Conn_Map[i] = pSt;
				}
			}
		}
		else
			break;
	}

	bool bret = true;
	for(int i= CONNMANAGER_MIN_PARTITION; i<= m_iPartition; i++)
	{
		if(pIA[i] != i)
		{
			cerr<<__FILE__<<":"<<__func__<<":"<<"key = "<<i<<" has not be defined!"<<endl;
			bret = false;
			break;
		}
	}

	if(bret)
		for(int i=CONNMANAGER_MIN_PARTITION; i<= m_iPartition; i++)
		{
			ConnStruct *p = (ConnStruct *) m_Conn_Map[i];
			if(p->pCon==NULL || !p->pCon->CheckBackup(CONNMANAGER_MIN_PARTITION, m_iPartition))
			{
				cerr<<i<<" 's backup is error !"<<endl;
				bret = false;
				break;
			}
		}
	if(bret)
	{
		DEBUG1("now, let's connect!");
		for(int i=CONNMANAGER_MIN_PARTITION; i<= m_iPartition; i++)
		{
			ConnStruct *p = (ConnStruct *) m_Conn_Map[i];
			if(p!= NULL && p->pCon != NULL )
			{
				DEBUG2("Yes, Connect to ", i);
				p->pCon->Connect();
			}
		}
	}

        delete []pIA;
	return bret;
}	

 ConnManager :: ~ConnManager( )
{	
	map<int, ConnStruct *> :: iterator iter;
	for(iter=m_Conn_Map.begin(); iter!= m_Conn_Map.end(); iter++)
	{
		ConnStruct *p = iter->second;
		if(p != NULL )
		{
#ifdef _THREAD_SAFE	
			pthread_mutex_destroy(&p->lock);
#endif	
			if(p->pCon != NULL)
				delete p->pCon;
			p->pCon = NULL;
			delete p;
			p = NULL;
		}
	}
	m_Conn_Map.clear();		
}

int ConnManager :: Proxy(int key, const char *data, unsigned dlen, char **rval, unsigned &wsize, unsigned &rlen, bool bReply)
{
/*
	if(m_bRestart)
	{
		m_iReqCount++;
		if(m_iReqCount >= m_iPartition*m_iRestart)
		{
			INFO<<"ReqCounts = "<< m_iReqCount<<": restart !" <<endl;
			exit(-1);
		}
	}
*/
	int idx = key % m_iPartition ;
	if(idx == 0)
		idx = m_iPartition;

	int iret = 0;
	_Lock(idx);
	ConnStruct *p = (ConnStruct *)m_Conn_Map[idx];
	if(p->pCon->getStatus( m_iRestart ))
		iret=  p->pCon->Proxy(data, dlen, rval, wsize, rlen, bReply);
	else
		iret = -1;
	_UnLock(idx);
	if(iret >= 0)
		return iret;

	/*
	 *二次执行的目的：一次执行的失败可能由于远程server端主动断开，这就要求netproxy重新连接！
	 *二次执行时，不再进行getStatus()函数，因为有时间戳问题！
	 */

	DEBUG1("====  Master is fail, maybe connection is lost,  so try again !");
	_Lock(idx);
	if(p->pCon->ReConnect())
		iret=  p->pCon->Proxy(data, dlen, rval, wsize, rlen, bReply);
	_UnLock(idx);
	if(iret >= 0)
		return iret;

	DEBUG1("=====  Master is fail again, so proxy to Slave now!  =====");
        int trycounts = p->pCon->GetBackLength();
        int trynum =  0;
	while(trynum < trycounts)
	{ 
		ConnStruct *p = (ConnStruct *)m_Conn_Map[idx];
		int idxb = p->pCon->GetNextBackup(trynum);
                DEBUG2(" getNextBackup ID = ", idxb);
		_Lock(idxb);
		p = (ConnStruct *)m_Conn_Map[idxb];
		if( p->pCon->getStatus( m_iRestart ))
			iret = p->pCon->Proxy(data, dlen, rval, wsize, rlen, bReply);
		_UnLock(idxb);
		if(iret >= 0)
			return iret;
     	        DEBUG1("=====  Slave is fail, try next!");
	/*
     	        DEBUG1("=====  Slave is fail, mybe connection is lost, try again!");
		_Lock(idxb);
		if(p->pCon->ReConnect())
			iret = p->pCon->Proxy(data, dlen, rval, wsize, rlen, bReply);
		_UnLock(idxb);
		if(iret >= 0)
			return iret;
        */
	}	
	if(iret < 0)
	{   DEBUG1(" ====  All is down, can't use !"); }
	return iret;
}

int ConnManager :: ResetStatus(const char *ip, int port)
{
      //  int isize = m_Conn_Map.size();
        for(int idx = CONNMANAGER_MIN_PARTITION; idx <= m_iPartition; idx++)
        {
                _Lock(idx);
                ConnStruct *p = (ConnStruct *)m_Conn_Map[idx];
                DEBUG4("check ", p->pCon->m_sHostIP.c_str(), ":", p->pCon->m_iPort);
                if(strcmp(p->pCon->m_sHostIP.c_str(), ip) == 0 && p->pCon->m_iPort == port)
                {
                        DEBUG5("Yes, set m_lLasttime = ", time(NULL),"(old =", p->pCon->m_lLasttime,"); m_bLive = true");
                        p->pCon->m_lLasttime = time(NULL);
                        p->pCon->m_bLive = true;
                }
                _UnLock(idx);
        }
        return 0;
}

void ConnManager ::_Lock(int idx)
{
#ifdef _THREAD_SAFE
	ConnStruct *p = (ConnStruct *)m_Conn_Map[idx];
	if(p!=NULL)
		pthread_mutex_lock(&p->lock);
#endif
}

void ConnManager ::_UnLock(int idx)
{
#ifdef _THREAD_SAFE
	ConnStruct *p = (ConnStruct *)m_Conn_Map[idx];
	if(p!=NULL)
		pthread_mutex_unlock(&p->lock);
#endif
}

void ConnManager :: trim(string& str)
{
   string::size_type pos = str.find_last_not_of(' ');
   if(pos != string::npos)
   {
      str.erase(pos + 1);
      pos = str.find_first_not_of(' ');
      if(pos != string::npos) str.erase(0, pos);
   }
   else str.erase(str.begin(), str.end());
}


